package org.apache.rocketmq.example.kang.consumer.sample;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class LitePullConsumerSubscribeTest {

    public static volatile boolean running = true;
    public static void main(String[] args) throws Exception {

        DefaultLitePullConsumer litePullConsumer = new
                DefaultLitePullConsumer(SampleConfig.CONSUMER_GROUP);
        litePullConsumer.setNamesrvAddr(SampleConfig.NAMESRV_ADDR);
        litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        litePullConsumer.subscribe(SampleConfig.TOPIC, "*");
        litePullConsumer.setAutoCommit(true); //该值默认为 true
        litePullConsumer.start();
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                doConsumeSomething(messageExts);
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
    private static void doConsumeSomething(List<MessageExt> messageExts) {
        // 真正的业务处理
        System.out.printf("消费消息条数 %s%n", messageExts.size());
    }

}
